#例【4-1】
import json
import time
import uuid

# 定义MCP协议的请求结构
class MCPRequest:
    def __init__(self, model, prompts, resources=None, tools=None, config=None):
        """
        初始化MCP请求对象。

        :param model: 调用的语言模型名称
        :param prompts: 提示词列表，包含多轮对话
        :param resources: 资源列表，包含外部数据或工具（可选）
        :param tools: 工具列表，定义可调用的外部工具（可选）
        :param config: 配置信息，包含推理参数（可选）
        """
        self.model = model
        self.prompts = prompts
        self.resources = resources if resources else []
        self.tools = tools if tools else []
        self.config = config if config else {}

    def to_dict(self):
        """
        将请求对象转换为字典格式，便于序列化。

        :return: 请求对象的字典表示
        """
        return {
            "model": self.model,
            "prompts": self.prompts,
            "resources": self.resources,
            "tools": self.tools,
            "config": self.config,
        }

    def to_json(self):
        """
        将请求对象转换为JSON格式的字符串。

        :return: 请求对象的JSON字符串表示
        """
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)

# 定义一个模拟的LLM模型接口
class MockLLM:
    def __init__(self):
        """
        初始化模拟的LLM模型。
        """
        pass

    def generate_response(self, mcp_request):
        """
        根据MCP请求生成模型的响应。

        :param mcp_request: MCPRequest对象，包含上下文信息
        :return: 模型生成的响应文本
        """
        # 模拟模型处理延迟
        time.sleep(1)
        # 简单地将最后一轮用户输入作为响应返回
        last_prompt = mcp_request.prompts[-1]
        if last_prompt["role"] == "user":
            return f"模型响应：收到用户输入 - '{last_prompt['content']}'"
        else:
            return "模型响应：未收到用户输入"

# 定义一个会话管理器，负责处理上下文的存储和检索
class SessionManager:
    def __init__(self):
        """
        初始化会话管理器，使用字典存储会话信息。
        """
        self.sessions = {}

    def get_session(self, session_id):
        """
        根据会话ID获取会话信息。

        :param session_id: 会话ID
        :return: 会话的提示词列表
        """
        return self.sessions.get(session_id, [])

    def update_session(self, session_id, prompt):
        """
        更新会话信息，添加新的提示词。

        :param session_id: 会话ID
        :param prompt: 新的提示词字典，包含角色和内容
        """
        if session_id not in self.sessions:
            self.sessions[session_id] = []
        self.sessions[session_id].append(prompt)

# 模拟一个对话流程
def main():
    # 初始化会话管理器和模拟的LLM模型
    session_manager = SessionManager()
    llm = MockLLM()

    # 定义会话ID
    session_id = str(uuid.uuid4())

    # 用户输入的对话轮次
    user_inputs = [
        "你好，你是谁？",
        "你能做些什么？",
        "请告诉我今天的天气。"
    ]

    # 模拟对话过程
    for user_input in user_inputs:
        # 获取当前会话的历史提示词
        history_prompts = session_manager.get_session(session_id)

        # 创建当前用户输入的提示词
        user_prompt = {"role": "user", "content": user_input}

        # 更新会话历史
        session_manager.update_session(session_id, user_prompt)

        # 构建MCP请求对象
        mcp_request = MCPRequest(
            model="MockLLM",
            prompts=history_prompts + [user_prompt]
        )

        # 将请求对象转换为JSON格式并打印
        request_json = mcp_request.to_json()
        print(f"发送给模型的请求：\n{request_json}")

        # 调用模型生成响应
        response = llm.generate_response(mcp_request)

        # 打印模型的响应
        print(f"模型的响应：{response}\n")

        # 将模型的响应作为助手的提示词更新到会话中
        assistant_prompt = {"role": "assistant", "content": response}



#例【4-2】
import json
import time
import uuid
from typing import List, Dict, Optional

class MCPRequest:
    def __init__(self, model: str, prompts: List[Dict[str, str]], resources: Optional[List[Dict]] = None, tools: Optional[List[Dict]] = None, config: Optional[Dict] = None):
        """
        初始化MCP请求对象。

        :param model: 使用的语言模型名称
        :param prompts: 提示词列表，包含角色和内容
        :param resources: 资源列表，可选，默认为None
        :param tools: 工具列表，可选，默认为None
        :param config: 配置信息，可选，默认为None
        """
        self.model = model
        self.prompts = prompts
        self.resources = resources if resources else []
        self.tools = tools if tools else []
        self.config = config if config else {}

    def to_dict(self) -> Dict:
        """
        将MCP请求对象转换为字典格式。

        :return: 字典格式的MCP请求对象
        """
        return {
            "model": self.model,
            "prompts": self.prompts,
            "resources": self.resources,
            "tools": self.tools,
            "config": self.config,
        }

    def to_json(self) -> str:
        """
        将MCP请求对象转换为JSON字符串格式。

        :return: JSON字符串格式的MCP请求对象
        """
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)

class MockLLM:
    def generate_response(self, mcp_request: MCPRequest) -> str:
        """
        模拟语言模型生成响应。

        :param mcp_request: MCP请求对象
        :return: 模拟的模型响应字符串
        """
        time.sleep(1)  # 模拟处理延迟
        last_prompt = mcp_request.prompts[-1]
        if last_prompt["role"] == "user":
            return f"模型响应：收到用户输入 - '{last_prompt['content']}'"
        else:
            return "模型响应：未收到用户输入"

class SessionManager:
    def __init__(self):
        """
        初始化会话管理器，使用字典存储会话信息。
        """
        self.sessions = {}

    def get_session(self, session_id: str) -> List[Dict[str, str]]:
        """
        根据会话ID获取会话的提示词列表。

        :param session_id: 会话ID
        :return: 提示词列表
        """
        return self.sessions.get(session_id, [])

    def update_session(self, session_id: str, prompt: Dict[str, str]):
        """
        更新会话，添加新的提示词。

        :param session_id: 会话ID
        :param prompt: 新的提示词，包含角色和内容
        """
        if session_id not in self.sessions:
            self.sessions[session_id] = []
        self.sessions[session_id].append(prompt)

def main():
    session_manager = SessionManager()
    llm = MockLLM()
    session_id = str(uuid.uuid4())

    system_message = {"role": "system", "content": "您正在与智能助手对话。"}

    user_inputs = [
        "你好，你是谁？",
        "你能做些什么？",
        "今天天气如何？"
    ]

    for user_input in user_inputs:
        history_prompts = session_manager.get_session(session_id)

        if not history_prompts:
            prompts = [system_message]
        else:
            prompts = history_prompts

        user_prompt = {"role": "user", "content": user_input}
        prompts.append(user_prompt)

        mcp_request = MCPRequest(
            model="MockLLM",
            prompts=prompts
        )

        print(f"发送给模型的请求：\n{mcp_request.to_json()}")

        response = llm.generate_response(mcp_request)
        print(f"模型的响应：{response}\n")

        assistant_prompt = {"role": "assistant", "content": response}
        session_manager.update_session(session_id, user_prompt)
        session_manager.update_session(session_id, assistant_prompt)

if __name__ == "__main__":
    main()



#例【4-3】
import json
import uuid
import time
from typing import List, Dict, Callable, Optional

# 定义一个基本的Prompt数据结构，包含role、content以及可预留的插槽列表
class Prompt:
    def __init__(self, role: str, content: str, slots: Optional[Dict[str, str]] = None):
        self.id = str(uuid.uuid4())
        self.role = role
        self.content = content
        self.slots = slots if slots is not None else {}  # 定义插槽，键为槽位名，值为填充值或空字符串

    def fill_slot(self, slot_name: str, content: str):
        """
        填充指定槽位的内容
        """
        if slot_name in self.slots:
            self.slots[slot_name] = content
        else:
            print(f"[WARN] 未发现槽位 {slot_name}，自动创建。")
            self.slots[slot_name] = content

    def get_merged_prompt(self) -> str:
        """
        根据预定义的槽位替换规则，将槽位内容填入原始内容模板中
        占位符格式采用{槽位名称}，若未填充则保留原样
        """
        merged = self.content
        for key, value in self.slots.items():
            placeholder = "{" + key + "}"
            merged = merged.replace(placeholder, value)
        return merged

    def to_dict(self):
        return {
            "id": self.id,
            "role": self.role,
            "content": self.content,
            "slots": self.slots
        }

# 定义一个上下文管理器，用于管理多轮对话中的Prompt链
class ContextManager:
    def __init__(self):
        self.prompts: List[Prompt] = []

    def add_prompt(self, prompt: Prompt):
        """
        添加一个新的Prompt到上下文中
        """
        self.prompts.append(prompt)

    def get_context(self) -> List[Dict]:
        """
        返回整个上下文的序列化字典列表
        """
        return [prompt.to_dict() for prompt in self.prompts]

    def merge_all_prompts(self) -> str:
        """
        将所有Prompt通过换行拼接为一个完整的上下文文本，确保每个Prompt都经过插槽填充
        """
        merged_prompts = [prompt.get_merged_prompt() for prompt in self.prompts]
        return "\n".join(merged_prompts)

# 定义一个插件模拟环境，用于动态填充插槽内容
def fill_weather_info() -> str:
    """
    模拟从外部API获取天气信息
    """
    # 模拟网络延迟
    time.sleep(1)
    return "今天晴转多云，最高温度30摄氏度"

def fill_user_name() -> str:
    """
    模拟获取当前用户名称
    """
    # 直接返回模拟用户信息
    return "张三"

# 模拟一个对话流程，利用Prompt插槽式语义填充设计构建上下文
def main():
    # 初始化上下文管理器
    context_manager = ContextManager()

    # 定义系统指令提示词，不需要填充插槽
    system_prompt = Prompt(
        role="system",
        content="你是一个招聘助手，负责分析候选人的简历。"
    )
    context_manager.add_prompt(system_prompt)

    # 定义用户输入Prompt，其中包含插槽占位符，用于动态填充用户姓名和天气信息
    # 使用占位符 {user_name} 和 {weather_info} 分别表示需要动态填充的内容
    user_prompt = Prompt(
        role="user",
        content="你好，{user_name}，请问今天的天气如何？",
        slots={"user_name": "", "weather_info": ""}
    )
    context_manager.add_prompt(user_prompt)

    # 定义助手回复Prompt，同样包含插槽占位符
    assistant_prompt = Prompt(
        role="assistant",
        content="根据系统数据，当前天气状况为：{weather_info}。请继续描述候选人信息。",
        slots={"weather_info": ""}
    )
    context_manager.add_prompt(assistant_prompt)

    # 动态填充插槽
    # 填充用户名称槽位
    user_prompt.fill_slot("user_name", fill_user_name())
    # 填充天气信息槽位（同时用于用户和助手的Prompt）
    weather = fill_weather_info()
    user_prompt.fill_slot("weather_info", weather)
    assistant_prompt.fill_slot("weather_info", weather)

    # 构建完整的上下文文本
    complete_context = context_manager.merge_all_prompts()
    
    # 模拟输出最终的上下文与合并结果
    print("=== 完整上下文输出 ===")
    print(complete_context)
    print("\n=== 上下文结构化数据 ===")
    print(json.dumps(context_manager.get_context(), indent=2, ensure_ascii=False))

if __name__ == "__main__":
    main()



#例【4-4】
import base64
import json
import uuid
import time
from datetime import datetime
from PIL import Image
import io

# 假设此模块中的部分函数和类来源于 MCP-Chinese-Getting-Started-Guide 项目
# 该示例模拟图像上下文数据的封装与映射过程

class ImageContext:
    """
    图像上下文类，用于封装原始图像数据与语义描述，
    并将其转化为符合MCP协议的结构化Prompt对象
    """
    def __init__(self, image_path: str, description: str):
        self.image_path = image_path    # 图像文件路径
        self.description = description  # 图像的语义描述信息
        self.id = str(uuid.uuid4())     # 唯一标识，用于上下文链引用
        self.timestamp = datetime.utcnow().isoformat()  # 图像封装时间

    def preprocess_image(self) -> bytes:
        """
        对图像进行预处理，例如尺寸调整与格式转换
        返回处理后的图像二进制数据
        """
        # 打开图像并调整为固定尺寸
        with Image.open(self.image_path) as img:
            # 将图像调整为标准尺寸(如256x256)
            img = img.resize((256, 256))
            # 将图像转换为JPEG格式并保存到内存缓冲区
            buffer = io.BytesIO()
            img.save(buffer, format="JPEG")
            return buffer.getvalue()

    def encode_image_to_base64(self) -> str:
        """
        将预处理后的图像二进制数据编码为Base64字符串
        """
        image_data = self.preprocess_image()
        # 使用Base64进行编码，并解码为utf-8字符串
        encoded_str = base64.b64encode(image_data).decode("utf-8")
        return encoded_str

    def to_prompt(self) -> dict:
        """
        将图像上下文转化为符合MCP协议的Prompt对象，
        包含角色、内容、状态及其他元数据
        """
        prompt = {
            "id": self.id,
            "role": "image",  # 图像上下文一般标识为image角色
            "content": self.encode_image_to_base64(),  # 图像数据以Base64形式封装
            "metadata": {
                "description": self.description,
                "timestamp": self.timestamp,
                "format": "jpeg",
                "width": 256,
                "height": 256
            },
            "status": "active"  # 表示该图像Prompt当前处于活跃状态
        }
        return prompt

def merge_context(prompts: list) -> dict:
    """
    将多个Prompt对象整合到一个统一的上下文请求结构中，
    用于传递给模型调用
    """
    context_request = {
        "model": "deepseek-chat",
        "prompts": prompts,
        "config": {
            "temperature": 0.6,
            "max_tokens": 512,
            "stream": False
        },
        "metadata": {
            "request_id": str(uuid.uuid4()),
            "caller": "mcp-image-module",
            "timestamp": datetime.utcnow().isoformat(),
            "env": "production"
        }
    }
    return context_request

def main():
    # 定义图像文件路径（假定当前目录下有一张sample.jpg图像）
    image_path = "sample.jpg"
    image_description = "一张展示自然风景的照片，包含蓝天、白云和绿草。"

    # 创建图像上下文对象
    img_context = ImageContext(image_path, image_description)

    # 将图像上下文转换为Prompt格式
    image_prompt = img_context.to_prompt()

    # 构建完整的上下文请求（此处仅包含一个图像Prompt，可扩展至多模态）
    mcp_request = merge_context([image_prompt])

    # 将请求转换为JSON格式并输出
    request_json = json.dumps(mcp_request, ensure_ascii=False, indent=2)
    print("=== 构建的MCP上下文请求 ===")
    print(request_json)

    # 模拟一个处理延迟，代表模型调用过程
    time.sleep(1)
    simulated_response = {
        "status": "success",
        "trace_id": str(uuid.uuid4()),
        "outputs": [
            {
                "role": "assistant",
                "content": "模型收到图像上下文，生成相关描述：这是一张展示自然风景的照片，整体色彩明快，视觉效果良好。"
            }
        ]
    }
    response_json = json.dumps(simulated_response, ensure_ascii=False, indent=2)
    print("\n=== 模拟模型响应 ===")
    print(response_json)

if __name__ == "__main__":
    main()



#例【4-5】
import json
import uuid
from datetime import datetime

# 定义表格数据的示例，使用列表表示表格的每一行，字典保存行数据
table_data = [
    {"姓名": "张三", "年龄": "28", "职位": "工程师", "工作经验": "5年"},
    {"姓名": "李四", "年龄": "32", "职位": "产品经理", "工作经验": "8年"},
    {"姓名": "王五", "年龄": "26", "职位": "设计师", "工作经验": "3年"}
]

# 定义表格Prompt模板，包含占位符
prompt_template = (
    "以下是一份候选人简历表格数据，格式为：\n"
    "列名：姓名、年龄、职位、工作经验。\n"
    "数据内容如下：\n"
    "{table_content}\n\n"
    "请根据以上数据分析各候选人与目标岗位要求的匹配度，并生成简要总结。"
)

def format_table_data(data: list) -> str:
    """
    将表格数据转换为文本格式，用于填充到Prompt模板中。
    
    :param data: 表格数据列表，每个元素是字典，键为字段名称，值为具体内容。
    :return: 格式化后的表格文本，每行数据以逗号分隔，行间以换行符区分。
    """
    if not data:
        return ""
    
    # 获取所有列标题，假设所有行的键相同
    columns = list(data[0].keys())
    # 构建标题行
    title_line = "、".join(columns)
    
    # 构建数据行，每一行将各字段值以逗号分隔
    row_lines = []
    for row in data:
        row_text = "、".join(row[col] for col in columns)
        row_lines.append(row_text)
    
    # 拼接标题与所有数据行，形成完整表格文本
    table_text = f"{title_line}\n" + "\n".join(row_lines)
    return table_text

def construct_prompt(template: str, table_text: str) -> str:
    """
    根据模板填充表格文本，构造最终Prompt。
    
    :param template: 包含占位符的Prompt模板，使用 {table_content} 作为占位符。
    :param table_text: 格式化后的表格文本。
    :return: 合成后的完整Prompt字符串。
    """
    prompt = template.replace("{table_content}", table_text)
    return prompt

def build_mcp_request(prompt_content: str) -> dict:
    """
    构建MCP请求对象，其中包含模型调用的Prompt链，及相关元数据设置。
    
    :param prompt_content: 合成后的Prompt文本。
    :return: MCP请求对象字典，符合MCP协议格式。
    """
    request_object = {
        "model": "deepseek-chat",
        "prompts": [
            {
                "role": "system",
                "content": "你是一个专业的招聘助手，专注于候选人匹配度分析。"
            },
            {
                "role": "user",
                "content": prompt_content
            }
        ],
        "config": {
            "temperature": 0.6,
            "max_tokens": 512,
            "stream": False
        },
        "metadata": {
            "request_id": str(uuid.uuid4()),
            "caller": "agent.hr.analysis",
            "timestamp": datetime.utcnow().isoformat(),
            "env": "production"
        }
    }
    return request_object

def simulate_model_response(mcp_request: dict) -> dict:
    """
    模拟大语言模型处理MCP请求后返回响应。
    
    :param mcp_request: MCP请求对象字典。
    :return: 模拟的响应对象字典。
    """
    # 模拟处理延迟
    import time
    time.sleep(1)
    # 模拟返回响应，由于此处为示例，响应直接使用固定模板填充用户Prompt的后半部分。
    response = {
        "status": "success",
        "trace_id": str(uuid.uuid4()),
        "outputs": [
            {
                "role": "assistant",
                "content": "根据表格数据，候选人张三的工程师背景与岗位要求较为匹配；李四在管理经验上具有优势，但与技术要求存在差距；王五虽设计能力突出，但工作经验不足。总体建议针对岗位需求进一步细化筛选标准。"
            }
        ]
    }
    return response

def main():
    # 将表格数据格式化为文本
    table_text = format_table_data(table_data)
    
    # 根据模板构造完整Prompt
    prompt_text = construct_prompt(prompt_template, table_text)
    
    # 打印合成后的Prompt内容
    print("=== 合成的Prompt ===")
    print(prompt_text)
    print("\n")
    
    # 构建MCP请求对象
    mcp_request = build_mcp_request(prompt_text)
    
    # 将请求对象转换为JSON格式并输出
    request_json = json.dumps(mcp_request, ensure_ascii=False, indent=2)
    print("=== 构建的MCP请求 ===")
    print(request_json)
    print("\n")
    
    # 模拟模型处理请求并生成响应
    response = simulate_model_response(mcp_request)
    
    # 输出模型响应
    response_json = json.dumps(response, ensure_ascii=False, indent=2)
    print("=== 模拟模型响应 ===")
    print(response_json)

if __name__ == "__main__":
    main()



#例【4-6】
import json
import uuid
from datetime import datetime

# 模拟文档预处理的功能，包括清洗和格式化
def preprocess_document(file_path: str) -> str:
    """
    读取文档，进行文本清洗，统一换行，并返回预处理后的文本内容。
    
    参数:
        file_path: 文档的文件路径
        
    返回:
        预处理后的文档文本字符串
    """
    try:
        with open(file_path, "r", encoding="utf-8") as file:
            raw_text = file.read()
    except Exception as e:
        raise Exception(f"读取文件失败: {e}")
    
    # 文本清洗示例：去除多余空格、统一换行符
    cleaned_text = raw_text.strip()
    # 替换多个连续空格为单个空格
    cleaned_text = " ".join(cleaned_text.split())
    # 替换回车换行为统一的换行符
    cleaned_text = cleaned_text.replace("\\n", "\n").replace("\r\n", "\n")
    
    return cleaned_text

# 定义一个函数用于提取文档摘要，模拟关键信息提取
def extract_summary(text: str) -> str:
    """
    简单模拟文档摘要的提取，截取前200个字符作为摘要。
    
    参数:
        text: 文档的预处理文本
        
    返回:
        文档摘要文本
    """
    summary_length = 200
    # 如果文档长度大于摘要长度，则截取摘要；否则返回全文
    if len(text) > summary_length:
        return text[:summary_length] + "..."
    return text

# 构造文档嵌入的上下文Prompt对象
def build_document_prompt(file_path: str, title: str) -> dict:
    """
    根据文档文件路径和标题生成对应的Prompt对象，
    包含预处理后的文档内容摘要以及相关元数据。
    
    参数:
        file_path: 文档文件的路径
        title: 文档标题
        
    返回:
        一个符合MCP标准的Prompt对象（字典形式）
    """
    # 对文档进行预处理
    preprocessed_text = preprocess_document(file_path)
    # 提取文档摘要
    summary = extract_summary(preprocessed_text)
    
    # 构造Prompt对象
    prompt = {
        "id": str(uuid.uuid4()),
        "role": "document",
        "content": summary,
        "metadata": {
            "title": title,
            "source": file_path,
            "created_at": datetime.utcnow().isoformat(),
            "summary": summary,
            "full_length": len(preprocessed_text)
        },
        "status": "active"
    }
    return prompt

# 构建MCP请求对象，将文档Prompt嵌入上下文
def build_mcp_request_with_document(document_prompt: dict) -> dict:
    """
    构建一个MCP请求对象，该请求包含文档嵌入的Prompt，
    用于将预处理后的文档数据注入至语言模型上下文中。
    
    参数:
        document_prompt: 经过封装的文档Prompt对象
        
    返回:
        完整的MCP请求对象
    """
    request_object = {
        "model": "deepseek-chat",
        "prompts": [
            {
                "role": "system",
                "content": "你是一个专业的文档摘要分析助手。"
            },
            document_prompt  # 将文档嵌入的Prompt添加进上下文
        ],
        "config": {
            "temperature": 0.5,
            "max_tokens": 512,
            "stream": False
        },
        "metadata": {
            "request_id": str(uuid.uuid4()),
            "caller": "doc-embed-module",
            "timestamp": datetime.utcnow().isoformat(),
            "env": "production",
            "task_type": "document_embedding"
        }
    }
    return request_object

def main():
    # 假设当前目录下有一个名为sample_doc.txt的文档
    file_path = "sample_doc.txt"
    document_title = "项目需求文档"
    
    # 构造文档Prompt对象
    document_prompt = build_document_prompt(file_path, document_title)
    
    # 构建包含文档Prompt的MCP请求对象
    mcp_request = build_mcp_request_with_document(document_prompt)
    
    # 将请求对象转换为JSON格式字符串并输出
    request_json = json.dumps(mcp_request, ensure_ascii=False, indent=2)
    print("=== 构建的MCP请求 ===")
    print(request_json)
    
    # 模拟模型处理过程，生成一个响应（实际中此处将调用模型接口）
    simulated_response = {
        "status": "success",
        "trace_id": str(uuid.uuid4()),
        "outputs": [
            {
                "role": "assistant",
                "content": "模型已处理文档摘要，生成相关分析：该文档主要描述了项目需求，内容详尽，逻辑清晰。"
            }
        ]
    }
    response_json = json.dumps(simulated_response, ensure_ascii=False, indent=2)
    print("\n=== 模拟模型响应 ===")
    print(response_json)

if __name__ == "__main__":
    main()



#例【4-7】
import time
import json
import uuid

# 模拟Token流生成器
def token_stream_generator():
    """
    模拟一个Token流生成器，每次迭代返回一部分Token。
    生成的Token不一定构成完整的语句，需要解码模块拼接。
    """
    tokens = [
        "您好", "，", "欢迎", "使用", "大语言", "模型", "实时", "响应", "系统", "。", 
        "本次", "生成", "将", "展示", "Token流", "中间态", "解码", "策略", "的", "效果", "。",
        "系统", "在", "不断", "接收", "Token", "的", "同时", "保持", "语义", "连贯", "，",
        "确保", "输出", "内容", "完整", "且", "及时", "呈现", "给", "用户", "。"
    ]
    for token in tokens:
        time.sleep(0.1)  # 模拟生成延迟
        yield token

class TokenDecoder:
    """
    Token解码器用于实时拼接Token流，
    并在遇到句号或指定结束符时输出当前完成的文本块。
    """
    def __init__(self, end_markers=None):
        # 设置结束标记，默认以句号为结束标记
        self.end_markers = end_markers if end_markers is not None else ["。"]
        self.buffer = ""

    def process_token(self, token: str) -> str:
        """
        将接收到的Token加入缓冲区，检测是否达到结束标记，
        如达到，则返回完整的语句，同时清空缓冲区。
        如果没有，则返回空字符串，表示未形成完整语句。

        :param token: 模拟生成的Token
        :return: 完整语句或空字符串
        """
        self.buffer += token
        # 检查缓冲区末尾是否含有任一结束标记
        for marker in self.end_markers:
            if self.buffer.endswith(marker):
                completed_sentence = self.buffer
                self.buffer = ""
                return completed_sentence
        # 如果未检测到结束标记，返回空字符串
        return ""

    def flush(self) -> str:
        """
        在Token流结束后，返回缓冲区内剩余内容，并清空缓冲区。
        :return: 缓冲区中剩余的内容
        """
        remaining = self.buffer
        self.buffer = ""
        return remaining

def simulate_stream_decoding():
    """
    模拟Token流实时解码过程，
    并输出每个完整语句。
    """
    decoder = TokenDecoder()
    full_output = []  # 用于存储所有完整输出的文本块
    print("=== 开始流式解码 ===")
    # 遍历Token流
    for token in token_stream_generator():
        # 处理每个Token并检查是否生成完整语句
        sentence = decoder.process_token(token)
        if sentence:
            # 当检测到完整语句时，输出并保存
            print("完整语句输出：", sentence)
            full_output.append(sentence)
    # 流结束后，检查缓冲区中是否有未输出部分
    remaining = decoder.flush()
    if remaining:
        print("剩余未完成语句：", remaining)
        full_output.append(remaining)
    print("\n=== 解码完成 ===")
    return full_output

if __name__ == "__main__":
    # 执行流式解码模拟
    output_sentences = simulate_stream_decoding()
    
    # 输出整体结果的JSON格式
    result = {
        "status": "success",
        "trace_id": str(uuid.uuid4()),
        "outputs": output_sentences
    }
    print("\n=== 模拟模型最终响应 ===")
    print(json.dumps(result, ensure_ascii=False, indent=2))



#例【4-8】
import json
import uuid
from datetime import datetime

# 定义一个基础的Prompt对象，模拟对话系统中的历史上下文单元
class Prompt:
    def __init__(self, role: str, content: str, timestamp: str = None):
        self.id = str(uuid.uuid4())
        self.role = role
        self.content = content
        self.timestamp = timestamp if timestamp else datetime.utcnow().isoformat()

    def to_dict(self):
        return {
            "id": self.id,
            "role": self.role,
            "content": self.content,
            "timestamp": self.timestamp
        }

# 定义一个ContextManager，用于管理历史Prompt
class ContextManager:
    def __init__(self):
        # 存储所有Prompt对象列表
        self.prompts = []
    
    def add_prompt(self, prompt: Prompt):
        # 添加一个新的Prompt到上下文中
        self.prompts.append(prompt)
    
    def get_recent_context(self, max_prompts=3) -> list:
        # 返回最近几条Prompt作为上下文提示注入依据
        return self.prompts[-max_prompts:] if len(self.prompts) >= max_prompts else self.prompts

# 定义Response对象，用于表示模型输出
class Response:
    def __init__(self, content: str):
        self.id = str(uuid.uuid4())
        self.content = content
        self.metadata = {}
    
    def inject_context_hint(self, context_text: str):
        # 注入上下文提示信息到响应内容中
        # 此处采用简单拼接方式，可根据实际需求设计模板格式
        self.content += "\n\n[上下文提示] " + context_text
    
    def to_dict(self):
        return {
            "id": self.id,
            "content": self.content,
            "metadata": self.metadata
        }

# 定义一个函数，用于合并历史Prompt成为上下文提示文本
def build_context_hint(prompts: list) -> str:
    """
    将一系列Prompt拼接为上下文提示文本，以供响应注入使用。
    
    参数:
        prompts: 提示词对象列表
    返回:
        拼接后的上下文提示字符串
    """
    context_lines = []
    for prompt in prompts:
        # 格式化每一条历史Prompt，包含角色与内容
        line = f"[{prompt.role}] {prompt.content}"
        context_lines.append(line)
    # 使用换行符连接所有提示
    return "\n".join(context_lines)

# 定义一个函数用于创建MCP请求对象并返回Response
def process_request(user_input: str, context_manager: ContextManager) -> Response:
    """
    处理单次请求：将用户输入加入上下文，调用模型生成响应，然后注入历史上下文提示。
    
    参数:
        user_input: 用户当前输入的文本
        context_manager: 当前对话上下文管理器实例
    返回:
        Response对象，包含生成的响应内容及注入上下文提示后文本
    """
    # 将当前用户输入封装为Prompt对象，并加入上下文管理器
    current_prompt = Prompt(role="user", content=user_input)
    context_manager.add_prompt(current_prompt)
    
    # 在实际场景中，此处调用大语言模型API获取响应，以下为固定响应文本
    response_content = f"模型响应：已处理用户输入 - '{user_input}'"
    response = Response(response_content)
    
    # 获取最近的上下文提示
    recent_prompts = context_manager.get_recent_context()
    context_hint = build_context_hint(recent_prompts)
    
    # 将上下文提示信息注入到响应内容中
    response.inject_context_hint(context_hint)
    
    # 将当前助手响应也加入上下文管理器作为历史记录
    assistant_prompt = Prompt(role="assistant", content=response.content)
    context_manager.add_prompt(assistant_prompt)
    
    return response

def main():
    # 初始化一个上下文管理器实例
    context_manager = ContextManager()
    
    # 添加初始系统Prompt
    system_prompt = Prompt(role="system", content="你是一个多轮对话的智能助手。")
    context_manager.add_prompt(system_prompt)
    
    # 用户输入的一系列问答轮次
    user_inputs = [
        "你好，请介绍一下公司的发展历程。",
        "最近有什么重要新闻发布吗？",
        "请问当前的股市情况如何？"
    ]
    
    # 处理每个用户输入，并输出对应响应
    for user_input in user_inputs:
        response = process_request(user_input, context_manager)
        print("=== 响应输出 ===")
        print(response.content)
        print("\n")

if __name__ == "__main__":
    main()



#例【4-9】
import json
import uuid
from datetime import datetime
import time

# 定义示例表格数据，使用列表存储，每一项为字典表示一行数据
table_data = [
    {"姓名": "张三", "年龄": "28", "职位": "软件工程师", "工作经验": "5年"},
    {"姓名": "李四", "年龄": "32", "职位": "产品经理", "工作经验": "8年"},
    {"姓名": "王五", "年龄": "26", "职位": "UI设计师", "工作经验": "3年"}
]

def format_table(data: list) -> str:
    """
    将表格数据转换为标准文本格式，用于构造Prompt文本。
    使用固定顺序排列列标题与行数据，并以换行符分隔每一行。
    
    参数:
      data: 表格数据列表，每个元素为字典表示一行
  
    返回:
      格式化后的表格文本字符串
    """
    if not data:
        return ""
    # 获取所有列标题，假设所有字典键均相同
    columns = list(data[0].keys())
    # 构造标题行
    header = "、".join(columns)
    # 构造数据行
    rows = []
    for row in data:
        # 按照标题顺序取值，元素间用“、”分隔
        row_text = "、".join(str(row[col]) for col in columns)
        rows.append(row_text)
    # 拼接标题行和数据行，用换行符分隔
    table_text = header + "\n" + "\n".join(rows)
    return table_text

def build_prompt_from_table(table_text: str, doc_title: str) -> str:
    """
    根据预定义模板构造包含表格数据的Prompt文本。
    模板中预留占位符用于填充表格内容和文档标题。
    
    参数:
      table_text: 格式化后的表格文本
      doc_title: 文档或表格的标题
  
    返回:
      完整的Prompt文本字符串
    """
    template = (
        f"文档标题：{doc_title}\n\n"
        "以下是候选人信息的表格数据，请仔细阅读各行数据，以便后续进行分析和决策：\n\n"
        "{table_content}\n\n"
        "请根据表格内容对候选人的资质进行详细评估，并生成总结报告。"
    )
    prompt = template.replace("{table_content}", table_text)
    return prompt

def build_mcp_request(prompt_text: str) -> dict:
    """
    构造符合MCP协议的请求对象，将Prompt文本嵌入Prompts字段中。
    
    参数:
      prompt_text: 完整的Prompt文本字符串
  
    返回:
      MCP请求对象的字典表示
    """
    request_obj = {
        "model": "deepseek-chat",
        "prompts": [
            {
                "role": "system",
                "content": "你是一位招聘助手，专注于候选人资质评估。"
            },
            {
                "role": "user",
                "content": prompt_text
            }
        ],
        "config": {
            "temperature": 0.6,
            "max_tokens": 512,
            "stream": False
        },
        "metadata": {
            "request_id": str(uuid.uuid4()),
            "caller": "agent.hr.analysis",
            "timestamp": datetime.utcnow().isoformat(),
            "env": "production",
            "task_type": "resume_assessment"
        }
    }
    return request_obj

def simulate_deepseek_inference(request_obj: dict) -> dict:
    """
    模拟Deepseek推理服务接口的处理过程。
    在实际部署中，该函数将调用Deepseek的API，此处返回固定响应。
    
    参数:
      request_obj: MCP请求对象字典
  
    返回:
      模拟的模型响应对象字典
    """
    # 此处等待1秒以模拟推理延时
    time.sleep(1)
    # 模拟返回响应信息，通常包括响应状态、trace_id以及outputs字段
    response = {
        "status": "success",
        "trace_id": str(uuid.uuid4()),
        "outputs": [
            {
                "role": "assistant",
                "content": "经过分析，候选人张三的技能和经验与岗位要求高度匹配；李四虽然管理经验丰富，但在技术能力上略逊；王五需要进一步培养设计与项目管理能力。建议企业重点考虑张三，另对李四进行技术能力补充考核。"
            }
        ]
    }
    return response

def main():
    # 定义一个文档标题
    doc_title = "候选人简历信息表"
    
    # 将示例表格数据格式化为文本字符串
    table_text = format_table(table_data)
    
    # 基于预处理后的表格文本构造完整的Prompt文本
    prompt_text = build_prompt_from_table(table_text, doc_title)
    
    # 构造MCP请求对象
    mcp_request = build_mcp_request(prompt_text)
    
    # 将请求对象转换为JSON并打印，便于查看结构
    request_json = json.dumps(mcp_request, ensure_ascii=False, indent=2)
    print("=== 构建的MCP请求 ===")
    print(request_json)
    
    # 调用Deepseek推理服务接口，获取响应
    response = simulate_deepseek_inference(mcp_request)
    
    # 将响应对象转换为JSON并输出
    response_json = json.dumps(response, ensure_ascii=False, indent=2)
    print("\n=== 模型响应 ===")
    print(response_json)

if __name__ == "__main__":
    main()



#例【4-10】
import json
import uuid
import time
from datetime import datetime

# 定义一个KVCache类，用于存储和管理Token流生成的中间状态
class KVCache:
    def __init__(self):
        # 使用字典结构存储缓存数据，键为层级编号，值为Token列表
        self.cache = {}

    def initialize_layer(self, layer: int):
        # 初始化指定层的缓存列表
        if layer not in self.cache:
            self.cache[layer] = []

    def add_tokens(self, layer: int, tokens: list):
        # 添加生成的Token到指定层的缓存中
        self.initialize_layer(layer)
        self.cache[layer].extend(tokens)

    def get_tokens(self, layer: int) -> list:
        # 返回指定层的缓存Token列表
        return self.cache.get(layer, [])

    def clear_cache(self, layer: int = None):
        # 清除缓存数据，可以清除指定层或全部清除
        if layer is not None:
            self.cache[layer] = []
        else:
            self.cache = {}

    def to_json(self) -> str:
        # 返回缓存数据的JSON格式字符串，便于传输和调试
        return json.dumps(self.cache, ensure_ascii=False, indent=2)

# 定义一个ContextAlignment类，用于实现上下文对齐策略
class ContextAlignment:
    def __init__(self, kv_cache: KVCache):
        # 依赖KVCache对象
        self.kv_cache = kv_cache

    def align_context(self, current_context: list, max_tokens: int) -> list:
        """
        根据当前上下文和最大Token数限制，对上下文Token流进行对齐和裁剪
        保证返回的Token列表长度不超过max_tokens，并优先保留最近生成的Token
        
        参数:
            current_context: 当前上下文Token列表
            max_tokens: 最大允许的Token数
        返回:
            对齐后的Token列表
        """
        # 如果当前Token数已满足要求，则返回完整上下文
        if len(current_context) <= max_tokens:
            return current_context
        # 否则，从后向前截取max_tokens个Token
        return current_context[-max_tokens:]

    def merge_with_cache(self, current_tokens: list, layer: int, max_tokens: int) -> list:
        """
        从KV Cache中获取指定层的Token，并与当前生成的Token列表合并，
        最终返回符合Token数量限制的上下文Token列表。
        
        参数:
            current_tokens: 当前生成的Token列表
            layer: KV Cache中使用的层级编号
            max_tokens: 最大Token总数限制
        返回:
            合并且对齐后的Token列表
        """
        # 从KV Cache获取历史Token
        cached_tokens = self.kv_cache.get_tokens(layer)
        # 合并缓存Token和当前生成Token
        combined_tokens = cached_tokens + current_tokens
        # 根据最大Token数限制进行裁剪，并返回结果
        aligned_tokens = self.align_context(combined_tokens, max_tokens)
        return aligned_tokens

# 定义一个函数，用于将Token列表转换为文本字符串（简单拼接）
def tokens_to_text(tokens: list) -> str:
    return "".join(tokens)

# 以下模拟从语言模型获得的Token流生成函数
def generate_token_stream() -> list:
    """
    生成一段Token流，每个Token为一个单词片段
    返回完整Token列表
    """
    tokens = [
        "现", "在", "开", "始", "对", "多", "轮", "下", "文", "的", "对", "齐", "处", "理", "。", 
        "通", "过", "快", "速", "Token", "流", "的", "缓存", "和", "上下", "文", "对", "齐", 
        "机", "制", "提", "高", "了", "生成", "效", "率", "，", "并", "保", "持", "了", "文", "本", 
        "语", "义", "的", "连", "贯", "性", "和", "精", "准", "度", "。", "系", "统", "根", "据", 
        "设", "定", "的", "Token", "数", "量", "上", "限", "，", "自", "动", "对", "现", "有", "上", 
        "下", "文", "进", "行", "裁", "剪", "，", "以", "确", "保", "后", "继", "输", "入", "的", 
        "完", "整", "性", "。"
    ]
    return tokens

def main():
    # 初始化KV Cache和ContextAlignment对象
    kv_cache = KVCache()
    context_aligner = ContextAlignment(kv_cache)
    
    # 定义一个层级编号，比如使用层1
    layer = 1
    
    # 清空KV Cache，为本次实验初始化
    kv_cache.clear_cache()
    
    # 预先将一些历史Token加入KV Cache，模拟之前生成的文本片段
    history_tokens = ["这", "是", "上", "一", "轮", "对", "话", "的", "Token", "数", "据", "，", "保", "持", "着", "语", "义", "连", "贯", "性", "。"]
    kv_cache.add_tokens(layer, history_tokens)
    
    # 从语言模型获得当前生成的Token流
    current_tokens = generate_token_stream()  # 返回Token列表
    
    # 定义Token总数限制，比如设置为50个Token
    max_tokens = 50
    
    # 使用ContextAlignment合并KV Cache中的历史Token和当前生成的Token，并进行对齐裁剪
    aligned_tokens = context_aligner.merge_with_cache(current_tokens, layer, max_tokens)
    
    # 将对齐后的Token列表转换为文本输出
    final_text = tokens_to_text(aligned_tokens)
    
    # 输出上下文KV Cache内容和最终对齐后的文本
    print("=== KV Cache 内容 ===")
    print(kv_cache.to_json())
    print("\n=== 对齐后的上下文文本 ===")
    print(final_text)

if __name__ == "__main__":
    main()